package com.dayu.tcp.service.server;

import com.dayu.tcp.common.Platform;
import com.dayu.tcp.count.CountHandler;
import com.dayu.tcp.count.CountInfo;
import com.dayu.tcp.listener.BeforeMessageEventListener;
import com.dayu.tcp.logging.Log;
import com.dayu.tcp.logging.LogFactory;
import com.dayu.tcp.protocol.codec.K1Decoder;
import com.dayu.tcp.protocol.codec.K1Encoder;
import com.dayu.tcp.service.EventDispatcher;
import com.dayu.tcp.service.Service;
import com.dayu.tcp.service.SocketType;
import com.dayu.tcp.service.WrappedChannel;
import com.dayu.tcp.service.status.StatusServer;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.epoll.EpollEventLoopGroup;
import io.netty.channel.epoll.EpollServerSocketChannel;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import lombok.Getter;
import lombok.Setter;

import java.net.InetSocketAddress;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author yehangtech
 * @date 2023/11/27 17:05
 * @description Socket Server
 */
public class Server extends Service {

    private static final Log log = LogFactory.getLog(Server.class);

    @Getter
    protected ConcurrentMap<String, WrappedChannel> channelMap = new ConcurrentHashMap<>();

    /**
     * 在握手协议中赋值
     * key->channel id。 value->AES加密字符串
     */
    protected ConcurrentMap<String, String> encryptKeyMap = new ConcurrentHashMap<>();

    /**
     * 服务端的名称
     */
    @Getter
    @Setter
    protected String serviceName;

    /**
     * Server 开始运行的时间
     */
    @Getter
    @Setter
    protected long startTime = System.currentTimeMillis();

    protected ServerBootstrap serverBootstrap;
    protected EventLoopGroup bossGroup;
    protected EventLoopGroup workerGroup;

    /**
     * Server统计信息
     */
    @Getter
    @Setter
    protected CountInfo countInfo = new CountInfo();

    /**
     * 是否开启统计信息
     */
    @Getter
    @Setter
    protected boolean openCount = true;

    /**
     * 统计处理器，注册到Server上
     */
    @Getter
    @Setter
    protected CountHandler countHandler;

    /**
     * 是否开启StatusServer
     */
    @Getter
    @Setter
    protected boolean openStatus = false;

    /**
     * 统计信息端口，默认8001
     */
    @Getter
    @Setter
    protected int statusPort = 8001;


    @Override
    public void shutdown() {
        if (null != bossGroup) {
            bossGroup.shutdownGracefully();
        }
        if (null != workerGroup) {
            workerGroup.shutdownGracefully();
        }
    }

    public Server() {
        super();
        //Server端默认使用业务处理线程池
        super.openExecutor = true;
    }

    @Override
    protected void init() {
        super.init();
        if (Platform.useEpoll()) {
            bossGroup = new EpollEventLoopGroup(workerCount, new ThreadFactory() {
                private final AtomicInteger index = new AtomicInteger(0);

                @Override
                public Thread newThread(Runnable runnable) {
                    return new Thread(runnable, "LINUX_BOSS_" + index.incrementAndGet());
                }
            });
            workerGroup = new EpollEventLoopGroup(workerCount, new ThreadFactory() {
                private final AtomicInteger index = new AtomicInteger(0);

                @Override
                public Thread newThread(Runnable runnable) {
                    return new Thread(runnable, "LINUX_WORK_" + index.incrementAndGet());
                }
            });
        } else {
            bossGroup = new NioEventLoopGroup(workerCount, new ThreadFactory() {
                private final AtomicInteger index = new AtomicInteger(0);

                @Override
                public Thread newThread(Runnable runnable) {
                    return new Thread(runnable, "BOSS_" + index.incrementAndGet());
                }
            });
            workerGroup = new NioEventLoopGroup(workerCount, new ThreadFactory() {
                private final AtomicInteger index = new AtomicInteger(0);

                @Override
                public Thread newThread(Runnable runnable) {
                    return new Thread(runnable, "WORK_" + index.incrementAndGet());
                }
            });
        }
        //初始化事件分发器
        eventDispatcher = new EventDispatcher(this);
        //框架层会第一时间处理消息事件
        this.addEventListener(new BeforeMessageEventListener());

        if (openCount) {
            countHandler = new CountHandler();
            ServerContext.getServerContext().setServer(this);
        }

    }

    public ChannelFuture bind() {
        init();
        serverBootstrap = new ServerBootstrap();
        serverBootstrap.group(bossGroup, workerGroup)
                .childOption(ChannelOption.SO_KEEPALIVE, super.keepAlive)
                .childOption(ChannelOption.TCP_NODELAY, super.tcpNoDelay)
                .channel(Platform.useEpoll() ? EpollServerSocketChannel.class : NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel socketChannel) {
                        ChannelPipeline pipeline = socketChannel.pipeline();
                        //1: 只有TCP模式下才选择去注册编解码器
                        if (Objects.equals(socketType, SocketType.NORMAL)) {
                            pipeline.addLast("K1Decoder", new K1Decoder());
                            pipeline.addLast("K1Encoder", K1Encoder.getSingleton());
                        }
                        //2: 注册各种自定义Handler
                        LinkedHashMap<String, ChannelHandler> channelHandlerMap = getChannelHandlerMap();
                        for (Map.Entry<String, ChannelHandler> entry : channelHandlerMap.entrySet()) {
                            pipeline.addLast(entry.getKey(), entry.getValue());
                        }
                        //3: 注册事件分发器
                        pipeline.addLast("ServerDispatchHandler", new ServerDispatchHandler(eventDispatcher));

                        //4: 注册Server统计信息处理器
                        if (openCount) {
                            pipeline.addLast("CountHandler", countHandler);
                        }
                    }
                });
        final InetSocketAddress socketAddress = new InetSocketAddress(port);
        ChannelFuture future = serverBootstrap.bind(socketAddress);
        //1：服务端与客户端建立链接后 boss线程会调用回调接口。第一个注册的是大禹-TCP 自定义的回调接口
        this.addFutureEventListener((ChannelFutureListener) ch -> {
            if (ch.isSuccess()) {
                log.info("Server started, listening on '" + socketAddress + "'");
            } else {
                log.error("Failed to start Server '" + socketAddress + "', caused by: ", ch.cause());
            }
        });
        //2：服务端与客户端建立链接后 boss线程会调用回调接口。然后才执行用户自定义的回调接口
        futureEventListenerList.forEach(future::addListener);

        //3：检查是否开启了状态服务器
        if (openStatus) {
            StatusServer statusServer = new StatusServer();
            statusServer.setPort(statusPort);
            statusServer.setKeepAlive(true);
            statusServer.setTcpNoDelay(true);
            statusServer.bind();
        }
        return future;
    }

    public WrappedChannel getChannel(String channelId) {
        if (channelId != null) {
            return channelMap.get(channelId);
        }
        return null;
    }
}
